package com.boot.mapreduce;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

public class Worker implements Runnable {
    private final Coordinator coordinator;
    private final int workerId;

    public Worker(Coordinator coordinator, int workerId) {
        this.coordinator = coordinator;
        this.workerId = workerId;
    }

    @Override
    public void run() {
        while (true) {
            Job.Task task = coordinator.getNextTask();
            if (task == null) break;
            try {
                if (task.type == Job.Task.Type.MAP) {
                    processMapTask(task);
                } else {
                    processReduceTask(task);
                }
            } catch (Exception e) {
                System.err.printf("Worker %d failed task %d: %s\n",
                workerId, task.taskId, e.getMessage());
            }
        }
    }

    public void processMapTask(Job.Task task) {
        // 模拟读取输入文件
        List<String> lines = Arrays.asList(
                "hello world",
                "hello java",
                "hello cpu"
        );
        //调用mapper
        List<Pair<String, Integer>> results = new ArrayList<>();
        Mapper<?, ?, ?, ?> rawMapper = coordinator.getJob().getMapper();
        @SuppressWarnings("unchecked")
        Mapper<Object,String, String, Integer> mapper =
                (Mapper<Object, String, String, Integer>) rawMapper;
        for (String line : lines) {
            results.addAll(mapper.map(null, line));
        }
        coordinator.storeIntermediateResult(task.taskId, results);
    }

    private void processReduceTask(Job.Task task) {
        // 模拟从中间结果读取
        List<Pair<String, List<Integer>>> inputs = Arrays.asList(
                new Pair<>("hello", Arrays.asList(6,3)),
                new Pair<>("world", Arrays.asList(2,3)),
                new Pair<>("java", Collections.singletonList(4))

        );
        // 调用用户Reducer
        Reducer<?, ?, ?, ?> rawReducer = coordinator.getJob().getReducer();
        @SuppressWarnings("unchecked")
        Reducer<String, Integer, String, Integer> reducer =
                (Reducer<String, Integer, String, Integer>) rawReducer;

        List<Pair<String, Integer>> results = new ArrayList<>();
        for (Pair<String, List<Integer>> input : inputs) {
            results.addAll(reducer.reduce(input.key, input.value));
        }

        coordinator.mergeFinalResult(task.taskId, results);
    }
}
